package com.xxx.datalake.controller.structuredData;

import com.xxx.datalake.service.GetMySqlResourcesData;
import com.xxx.datalake.service.HBaseConnection;
import com.xxx.datalake.service.JdbcConnection;
import com.xxx.datalake.service.RunDataX;
import com.xxx.datalake.utils.json.CreateJson;
import org.apache.hadoop.hbase.client.Admin;
import org.json.JSONException;
import org.springframework.stereotype.Controller;
import org.springframework.ui.Model;
import org.springframework.web.bind.annotation.*;

import javax.servlet.http.HttpSession;
import java.io.IOException;
import java.sql.Connection;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;

/**
 * @ Project: datalake
 * @ Since: 2022/9/19 10:45
 * @ Description:
 */
@Controller
@RequestMapping({"/data_lake"})
public class StructuredController {

    /**
     * 首页
     * @return
     */
    @GetMapping({"","/"})
    public String index() {
        return "index";
    }

    /**
     * 数据源页面
     * @return
     */
    @GetMapping({"/data_source"})
    public String sourceList() {
        return "data-source";
    }

    /**
     * 新增mysql数据源
     * @param jdbcUrl
     * @param jdbcUsername
     * @param jdbcPassword
     * @param session
     * @return
     */
    @PostMapping({"/data_source_mysql"})
    public String addMysqlSource(String jdbcUrl,
                            String jdbcUsername,
                            String jdbcPassword,
                            HttpSession session) {
        session.setAttribute("jdbcUrl",jdbcUrl);
        session.setAttribute("jdbcUsername",jdbcUsername);
        session.setAttribute("jdbcPassword",jdbcPassword);
        JdbcConnection connection = new JdbcConnection();
        Connection mySqlConnection = connection.getMySqlConnection(jdbcUrl, jdbcUsername, jdbcPassword);
        session.setAttribute("mySqlConnection",mySqlConnection);
        //连接测试，返回连接测试结果
        session.setAttribute("mysqlConn",true);//用来判断当前是否有mysql数据库信息
        return "redirect:/data_lake/data_source";
    }

    /**
     * 新增hbase数据源
     * @param hbaseUrl
     * @param session
     * @return
     */
    @PostMapping({"/data_source_hbase"})
    public String addHbaseSource(String hbaseUrl,
                                 HttpSession session) throws IOException {
        session.setAttribute("hbaseUrl",hbaseUrl);
        session.setAttribute("hbaseConn",true);//用来判断当前是否有hbase数据库信息
        HBaseConnection hBaseConnection = new HBaseConnection();
        Admin admin = hBaseConnection.getHBaseConnection(hbaseUrl);
        session.setAttribute("hbaseAdmin",admin);
        return "redirect:/data_lake/data_source";
    }

    /**
     * 移除相关数据源，根据请求的type类型移除指定的数据源
     * @param type
     * @param session
     * @return
     */
    @GetMapping({"/remove_data_source/{type}"})
    public String closeConnection(@PathVariable("type") String type,
                                  HttpSession session) {
        if (type.equals("mysql")){
            session.setAttribute("jdbcUrl",null);
            session.setAttribute("jdbcUsername",null);
            session.setAttribute("jdbcPassword",null);
            session.setAttribute("mysqlConn",false);
        }
        if (type.equals("hbase")){
            session.setAttribute("hbaseUrl",null);
            session.setAttribute("hbaseConn",false);
        }
        return "redirect:/data_lake/data_source";
    }

    /**
     * 结构化数据入湖第一步，构建json文件
     * @param model
     * @return
     */
    @GetMapping({"/selection_mysql"})
    public String createJson(Model model,HttpSession session) throws SQLException {
        //根据作用域的mysql连接信息，获取该库的所有表格的信息，以map形式返回
        GetMySqlResourcesData mySqlResourcesData = new GetMySqlResourcesData();
        Connection connection = (Connection)session.getAttribute("mySqlConnection");
        Map tables = mySqlResourcesData.getMySqlTables(connection);
        Set<String> keySet = tables.keySet();
        //在model中添加，map，以及该map的所有key（即所有表名）
        model.addAttribute("tables",tables);
        for (String key : keySet) {
            //在model中添加每个表的各自map
            model.addAttribute(key, tables.get(key));
        }
        return "selection-mysql";
    }

    @PostMapping({"/selection_hbase"})
    public String createJsonHbase(String mysqlTableName,
                                  String[] fieldsName,
                                  Model model,
                                  HttpSession session) {
        model.addAttribute("mysqlTableName",mysqlTableName);
        session.setAttribute("mysqlTableName",mysqlTableName);
        model.addAttribute("fieldsName",fieldsName);
        return "selection-hbase";
    }


    @PostMapping({"/create_json"})
    public String fieldSelection(String hbaseTableName,
                                 String hbaseColumnFamily,
                                 String[] mysqlField,
                                 String[] hbaseField,
                                 String isRowkey,
                                 HttpSession session) throws JSONException {
        HashMap<String, String> fieldMap = new HashMap<>();
        for (int i = 0; i < mysqlField.length; i++) {
            fieldMap.put(mysqlField[i],hbaseField[i]);
        }

        //调用生成json文件代码
        String filename = "testMysql2Hbase1";
        CreateJson createJson = new CreateJson();
        String mysql2HbaseJson = createJson.Mysql2HbaseJson(fieldMap, hbaseTableName, hbaseColumnFamily, isRowkey, filename, session);

        //将json文件设置到session中
        session.setAttribute("mysql2HbaseJson",mysql2HbaseJson);
        session.setAttribute("hbaseTableName",hbaseTableName);
        session.setAttribute("hbaseColumnFamily",hbaseColumnFamily);


        return "json-preview";
    }

    @ResponseBody
    @GetMapping({"/execute_json"})
    public String executeJson(HttpSession session) throws IOException {
        //执行json
        String jsonFile = (String) session.getAttribute("mysql2HbaseJson");
        String hbaseTableName = (String) session.getAttribute("hbaseTableName");
        Admin admin = (Admin) session.getAttribute("hbaseAdmin");
        String hbaseColumnFamily = (String) session.getAttribute("hbaseColumnFamily");
        RunDataX runDataX = new RunDataX();
        runDataX.runMySql2HBse(admin,hbaseTableName,hbaseColumnFamily,jsonFile);
        return "任务执行中，请稍后查看结果";
    }
}
